/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2024-2024. All rights reserved.
 * Description: 集合通信hcom头文件
 * Author: gaozhenyue
 * Create: 2024-03-25
 */

#include <string>
#include <hccl/hccl_types.h>
#include "hccl/hcom.h"
#include "hccl_impl_base.h"
#include "hccl_comm_cpu.h"
#include "hcom_cpu_private.h"
#include "config.h"
#include "param_check_pub.h"
#include "externalinput.h"
#include "rank_consistentcy_checker.h"

using namespace std;
using namespace hccl;

constexpr u32 SINGLE_SERVER_NUM_CPU = 1;
constexpr u32 LAST_FOUR_BYTES = 4; // 取最后四字节
constexpr s32 NEVER_TIME_OUT = -1; // 永不超时时间

struct HcpuRequestInfo {
    std::string opTag;
    std::string group;
    HcomOpDesc opDesc;
    HcpuRequestInfo(const std::string &opTag, const std::string &group, const HcomOpDesc &opDesc)
        : opTag(opTag), group(group), opDesc(opDesc)
    {}
};

inline void GenerateHcpuSendRecvOpTag(
    HcomOperationType opType, const char *group, u32 tag, u32 selfRank, u32 peerRank, std::string &opTag)
{
    std::string groupStr = (group == nullptr) ? HCCL_WORLD_GROUP : group;
    if (opType == HCOM_OP_TYPE_SEND) {
        opTag = groupStr + "_" + std::to_string(tag) + "_" + std::to_string(selfRank) + "_" + std::to_string(peerRank);
    } else if (opType == HCOM_OP_TYPE_RECV) {
        opTag = groupStr + "_" + std::to_string(tag) + "_" + std::to_string(peerRank) + "_" + std::to_string(selfRank);
    }
    return;
}


HcclResult HcomGroupRef(const char *group)
{
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    if (group == nullptr) {
        return HCCL_SUCCESS; // 全局通信域不需要查询flag
    }
    std::string strGroup = group;
    std::unique_lock<std::mutex> groupParaLock(hcomInfo.groupParamsLock);
    auto iter = hcomInfo.hcomGroupMap.find(strGroup);
    if (iter == hcomInfo.hcomGroupMap.end()) {
        HCCL_WARNING("[Get][CommByGroup]errNo[0x%016llx] group[%s] group is not exist",
            HCOM_ERROR_CODE(HCCL_E_AGAIN), strGroup.c_str());
        return HCCL_E_AGAIN;  // 不存在该服务器内相关dev的对应信息
    }
    iter->second.refCounter++;
    return HCCL_SUCCESS;
}

HcclResult HcomGroupUnref(const char *group)
{
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    if (group == nullptr) {
        return HCCL_SUCCESS; // 全局通信域不需要查询flag
    }
    std::string strGroup = group;
    std::unique_lock<std::mutex> groupParaLock(hcomInfo.groupParamsLock);
    auto iter = hcomInfo.hcomGroupMap.find(strGroup);
    if (iter == hcomInfo.hcomGroupMap.end()) {
        HCCL_WARNING("[Get][CommByGroup]errNo[0x%016llx] group[%s] group is not exist",
            HCOM_ERROR_CODE(HCCL_E_AGAIN), strGroup.c_str());
        return HCCL_E_AGAIN;  // 不存在该服务器内相关dev的对应信息
    }
    if (iter->second.refCounter > 0) {
        iter->second.refCounter--;
    }

    HCCL_INFO("iter->second.refCounter[%d]", iter->second.refCounter);
    return HCCL_SUCCESS;
}

HcclResult HcclCommGraphGetIdentifierCpu(s64 opBaseHcom, std::string &identifier)
{
    HcclCommCpu *hcclComm = reinterpret_cast<HcclCommCpu *>(opBaseHcom);
    identifier = hcclComm->GetIdentifier();
    return HCCL_SUCCESS;
}

HcclResult GenerateCclOpTagCpu(const std::string &opType, int64_t &hcomComm, std::string& group, std::string &sTag)
{
    HcomOpTagInfo &opTagInfo = HcomGetCtxOpTagInfo();

    // middle 在获取到hcomComm时，等于identifier；在获取到group时，等于group
    std::string middle;
    if (hcomComm == static_cast<int64_t>(CommNumHcom::COMM_VALUE_DEFAULT)) {
        middle = group;
    } else {
        CHK_RET(HcclCommGraphGetIdentifierCpu(hcomComm, middle));
    }
    std::hash<std::string> hashString;
    size_t nameHash = hashString(middle);
    sTag = opType + "_" + std::to_string(nameHash);
    // 多张图node name重复导致tag相同，因此对tag添加索引 tag = op type + node name + identifier name + index

    auto iter = opTagInfo.opIndex.find(middle);
    if (iter == opTagInfo.opIndex.end()) {
        opTagInfo.opIndex.insert({ middle, 0 });
        iter = opTagInfo.opIndex.find(middle);
        CHK_PRT_RET((iter == opTagInfo.opIndex.end()),
            HCCL_ERROR("[Generate][OpTag]generate tag fail. get the op index failed. ret[%d]", HCCL_E_INTERNAL),
            HCCL_E_INTERNAL);
        }

    sTag = sTag + "_" + std::to_string(iter->second++);

    HCCL_INFO("generate ccl op tag success, tag[%s]", sTag.c_str());
    return HCCL_SUCCESS;
}

HcclResult HcpuGetCommByGroup(const char *group, std::shared_ptr<HcclCommBase> &hcclComm)
{
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    CHK_RET(HcomCheckGroupName(group));
    std::string strGroup = (group == nullptr) ? HCCL_WORLD_GROUP : group;
    if (strGroup == HCCL_WORLD_GROUP) {
        hcclComm = hcomInfo.pCommBase;
        CHK_PRT_RET(
            hcomInfo.pCommBase == nullptr, HCCL_ERROR("[Get][CommByGroup]hcomInfo.pCommBase is null"), HCCL_E_PTR);
    } else {
        std::lock_guard<std::mutex> groupParaLock(hcomInfo.groupParamsLock);
        auto iter = hcomInfo.hcomGroupMap.find(strGroup);
        CHK_PRT_RET(iter == hcomInfo.hcomGroupMap.end(), HCCL_WARNING("[Get][CommByGroup]errNo[0x%016llx] group[%s]"
            "group is not exist", HCOM_ERROR_CODE(HCCL_E_NOT_FOUND), strGroup.c_str()), HCCL_E_NOT_FOUND);
        hcclComm = iter->second.pSubCommBase;
        CHK_PRT_RET(hcclComm == nullptr, HCCL_ERROR("[Get][CommByGroup] Get Comm is null"), HCCL_E_PTR);
    }
    return HCCL_SUCCESS;
}

HcclResult PrepareP2p(const HcomOpDesc *op, HcomRequest *requestHandle)
{
    HcclUs startut = TIME_NOW();
    CHK_PRT_RET((op->opType != HCOM_OP_TYPE_SEND) && (op->opType != HCOM_OP_TYPE_RECV),
        HCCL_ERROR("[Start][HcomPrepare]opType[%u] is invalid.", op->opType), HCCL_E_PARA);
    CHK_PRT_RET((op->schedType != HCOM_SCHED_TYPE_OS),
        HCCL_ERROR("[Start][HcomPrepare]schedType[%u] is invalid.", op->schedType), HCCL_E_PARA);
    CHK_PRT_RET((op->info.p2p.addr == nullptr) && (op->info.p2p.count != 0),
        HCCL_ERROR("[Start][HcomPrepare]addr[%p] is nullptr, count[%llu] is not zero.", op->info.p2p.addr,
        op->info.p2p.count), HCCL_E_PARA);
    *requestHandle = nullptr;
    std::shared_ptr<hccl::HcclCommBase> hcclComm;
    CHK_RET(HcpuGetCommByGroup(op->info.p2p.group, hcclComm));

    u32 selfRank = 0;
    CHK_RET(GetGroupRankInfo(op->info.p2p.group, RankInfoType::RANK_ID_IN_GROUP, 0, &selfRank));
    std::string opTag;
    GenerateHcpuSendRecvOpTag(op->opType, op->info.p2p.group, op->info.p2p.tag, selfRank, op->info.p2p.peerRank, opTag);

    CHK_RET(HcomCheckOpParam(opTag.c_str(), op->info.p2p.count, op->info.p2p.dataType));
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    CHK_RET(HcomCheckUserRank(hcomInfo.params.totalRanks, op->info.p2p.peerRank));

    /* 接口交互信息日志 */
    HCCL_RUN_INFO("Entry-HcomPrepareStart(PrepareP2p):opType[%u], tag[%u], addr[%p], count[%llu]," \
        " dataType[%s], peerRank[%u], opTag[%s], group[%s]", op->opType, op->info.p2p.tag, op->info.p2p.addr,
        op->info.p2p.count, GetDataTypeEnumStr(op->info.p2p.dataType).c_str(), op->info.p2p.peerRank,
        opTag.c_str(), op->info.p2p.group);

    /* 记录指令信息用于一致性校验 */
    HcclCMDType cmdType = (op->opType == HCOM_OP_TYPE_SEND) ?
        HcclCMDType::HCCL_CMD_SEND : HcclCMDType::HCCL_CMD_RECEIVE;
    CHK_RET(RankConsistentcyChecker::GetInstance().RecordOpPara(cmdType, opTag.c_str(), op->info.p2p.count,
        op->info.p2p.dataType, op->info.p2p.peerRank, op->info.p2p.tag, selfRank,
        hcclComm->inCCLbufferSize_, hcclComm->outCCLbufferSize_, op->info.p2p.group, hcclComm->GetRankTableCrc()));

    HcclResult ret = (reinterpret_cast<hccl::HcclCommCpu *>(hcclComm.get()))->PrepareStart(opTag,
        op->info.p2p.addr, op->info.p2p.count, op->info.p2p.dataType, op->info.p2p.peerRank);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[Start][HcomPrepare] hcclComm prepare start error, tag[%s], addr[%p], count[%llu], dataType[%s]"
        ", peerRank[%u], group[%s]", opTag.c_str(), op->info.p2p.addr, op->info.p2p.count,
        GetDataTypeEnumStr(op->info.p2p.dataType).c_str(), op->info.p2p.peerRank, op->info.p2p.group), ret);

    std::string groupStr = (op->info.p2p.group == nullptr) ? HCCL_WORLD_GROUP : op->info.p2p.group;
    *requestHandle = new (std::nothrow) HcomRequestInfo(opTag, groupStr, *op);
    CHK_PTR_NULL(*requestHandle);

    /* 关键状态记录 */
    HCCL_RUN_INFO("HcomPrepareStart(PrepareP2p) success, cost time %lld us. opType[%u], tag[%u], addr[%p]," \
        " count[%llu], dataType[%s], peerRank[%u], opTag[%s], group[%s], requestHandle[%p]",
        DURATION_US(TIME_NOW() - startut), op->opType,
        op->info.p2p.tag, op->info.p2p.addr, op->info.p2p.count, GetDataTypeEnumStr(op->info.p2p.dataType).c_str(),
        op->info.p2p.peerRank, opTag.c_str(), groupStr.c_str(), *requestHandle);

    return HCCL_SUCCESS;
}

HcclResult PrepareColl(const HcomOpDesc *op, HcomRequest *requestHandle)
{
    HcclUs startut = TIME_NOW();
    CHK_PRT_RET(op->opType >= HCOM_OP_TYPE_NUM, HCCL_ERROR("[Start][HcomPrepare]opType[%u] is invalid.", op->opType),
        HCCL_E_PARA);
    CHK_PRT_RET((op->schedType != HCOM_SCHED_TYPE_OS),
        HCCL_ERROR("[Start][HcomPrepare]schedType[%u] is invalid.", op->schedType), HCCL_E_PARA);
    CHK_PRT_RET((op->info.coll.inputAddr == nullptr) && (op->info.coll.count != 0),
        HCCL_ERROR("[Start][HcomPrepare]addr[%p] is nullptr, count[%llu] is not zero.", op->info.coll.inputAddr,
        op->info.coll.count), HCCL_E_PARA);

    std::shared_ptr<hccl::HcclCommBase> hcclComm;
    CHK_RET(HcpuGetCommByGroup(op->info.coll.group, hcclComm));

    std::string strGroup = (op->info.coll.group == nullptr) ? HCCL_WORLD_GROUP : op->info.coll.group;
    std::string sTag;
    int64_t hcclCommId = 0;
    if (op->opType == HCOM_OP_TYPE_GATHER) {
        CHK_RET(GenerateCclOpTagCpu(HCCL_KERNEL_OP_TYPE_GATHER, hcclCommId, strGroup, sTag));
        CHK_RET(RankConsistentcyChecker::GetInstance().RecordOpPara(HcclCMDType::HCCL_CMD_GATHER, sTag.c_str(),
            op->info.coll.count, op->info.coll.dataType, op->info.coll.root, hcclComm->inCCLbufferSize_,
            strGroup.c_str(), hcclComm->GetRankTableCrc()));
    } else if (op->opType == HCOM_OP_TYPE_BROADCAST) {
        CHK_RET(GenerateCclOpTagCpu(HCCL_KERNEL_OP_TYPE_BROADCAST, hcclCommId, strGroup, sTag));
        CHK_RET(RankConsistentcyChecker::GetInstance().RecordOpPara(HcclCMDType::HCCL_CMD_BROADCAST, sTag.c_str(),
            op->info.coll.count, op->info.coll.dataType, op->info.coll.root, hcclComm->inCCLbufferSize_, 0,
            strGroup.c_str(), hcclComm->GetRankTableCrc()));
    }

    /* 接口交互信息日志 */
    HCCL_RUN_INFO("Entry-HcomPrepareStart(PrepareColl):opType[%u], tag[%s], inputAddr[%p], outputAddr[%p], " \
        "count[%llu], dataType[%s], group[%s]", op->opType, sTag.c_str(), op->info.coll.inputAddr,
        op->info.coll.outputAddr, op->info.coll.count, GetDataTypeEnumStr(op->info.coll.dataType).c_str(),
        strGroup.c_str());

    HcclResult ret = (reinterpret_cast<hccl::HcclCommCpu *>(hcclComm.get()))->PrepareStartColl(sTag,
        op->info.coll, op->opType);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[Start][PrepareColl] hcclComm prepare start coll error, tag[%s], inputAddr[%p], count[%llu]," \
        " dataType[%s], group[%s]", sTag.c_str(), op->info.coll.inputAddr, op->info.coll.count,
        GetDataTypeEnumStr(op->info.coll.dataType).c_str(), op->info.coll.group), ret);
    *requestHandle = nullptr;

    HCCL_RUN_INFO("HcomPrepareStart(PrepareColl) success, cost time %lld us. opType[%u], tag[%s], inputAddr[%p]," \
        " outputAddr[%p], count[%llu], dataType[%s], group[%s]",
        DURATION_US(TIME_NOW() - startut), op->opType, sTag.c_str(), op->info.coll.inputAddr,
        op->info.coll.outputAddr, op->info.coll.count, GetDataTypeEnumStr(op->info.coll.dataType).c_str(),
        strGroup.c_str());
    return HCCL_SUCCESS;
}

HcclResult HcomPrepareStart(const HcomOpDesc *op, HcomRequest *requestHandle)
{
    // 入参合法性校验
    CHK_PTR_NULL(op);
    CHK_PTR_NULL(requestHandle);

    if (op->opType == HCOM_OP_TYPE_SEND || op->opType == HCOM_OP_TYPE_RECV) {
        CHK_RET(PrepareP2p(op, requestHandle));
    } else if (op->opType == HCOM_OP_TYPE_GATHER || op->opType == HCOM_OP_TYPE_BROADCAST) {
        CHK_RET(PrepareColl(op, requestHandle));
    }
    return HCCL_SUCCESS;
}

HcclResult HcomPrepareQuery(HcomRequest requestHandle, HcomStatus* status)
{
    // 入参合法性校验
    CHK_PTR_NULL(status);
    if (requestHandle == nullptr) {
        status->status = 0;
        HCCL_INFO("HcomPrepareQuery request handle is nullptr. return success");
        return HCCL_SUCCESS;
    }
    HcclUs startut = TIME_NOW();

    HcomRequestInfo* request = reinterpret_cast<HcomRequestInfo *>(requestHandle);

    HCCL_RUN_INFO("Entry-HcomPrepareQuery:requestHandle[%p] opTag[%s], group[%s]",
        requestHandle, request->opTag.c_str(), request->group.c_str());

    std::shared_ptr<hccl::HcclCommBase> hcclComm;
    CHK_RET(HcpuGetCommByGroup(request->group.c_str(), hcclComm));
    HcclResult ret = (reinterpret_cast<hccl::HcclCommCpu *>(hcclComm.get()))->PrepareQuerry(request->opTag,
        status->status);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[Querry][HcomPrepare] hcclComm prepare querry error, requestHandle[%p] opTag[%s], group[%s]",
        requestHandle, request->opTag.c_str(), request->group.c_str()), ret);

    /* 关键状态记录 */
    HCCL_RUN_INFO("HcomPrepareQuery success, cost time %lld us. requestHandle[%p], opTag[%s], group[%s], status[%u]",
        DURATION_US(TIME_NOW() - startut), requestHandle, request->opTag.c_str(), request->group.c_str(),
        status->status);

    if (status->status == 0) {
        RankConsistentcyChecker::GetInstance().DelOpPara(request->opTag);
        delete request;
    }
    return HCCL_SUCCESS;
}

HcclResult HcomPrepareCancel(HcomRequest requestHandle, HcomStatus* status)
{
    // 入参合法性校验
    CHK_PTR_NULL(status);

    if (requestHandle == nullptr) {
        status->status = 0;
        HCCL_INFO("HcomPrepareCancel request handle is nullptr. return success");
        return HCCL_SUCCESS;
    }
    HcclUs startut = TIME_NOW();

    HcomRequestInfo* request = reinterpret_cast<HcomRequestInfo *>(requestHandle);

    HCCL_RUN_INFO("Entry-HcomPrepareCancel:requestHandle[%p] opTag[%s], group[%s]",
        requestHandle, request->opTag.c_str(), request->group.c_str());

    std::shared_ptr<hccl::HcclCommBase> hcclComm;
    CHK_RET(HcpuGetCommByGroup(request->group.c_str(), hcclComm));
    HcclResult ret = (reinterpret_cast<hccl::HcclCommCpu *>(hcclComm.get()))->
        PrepareCancel(request->opTag, status->status);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[Cancel][HcomPrepare] hcclComm prepare cancel error, requestHandle[%p] opTag[%s], group[%s]",
        requestHandle, request->opTag.c_str(), request->group.c_str()), ret);
    RankConsistentcyChecker::GetInstance().DelOpPara(request->opTag);

    /* 关键状态记录 */
    HCCL_RUN_INFO("HcomPrepareCancel success, cost time %lld us. requestHandle[%p], opTag[%s], group[%s], status[%u]",
        DURATION_US(TIME_NOW() - startut), requestHandle, request->opTag.c_str(), request->group.c_str(),
        status->status);

    delete [] request;
    return HCCL_SUCCESS;
}

HcclResult HcomSendByOS(void *addr, uint64_t count, HcclDataType dataType, uint32_t peerRank, uint32_t tag,
    const char *group, uint64_t flag)
{
    HcclUs startut = TIME_NOW();
    CHK_PRT_RET(count == 0, HCCL_WARNING("input count is 0, return send success"), HCCL_SUCCESS);
    CHK_PTR_NULL(addr);

    // 入参合法性校验
    std::shared_ptr<hccl::HcclCommBase> hcclComm;
    CHK_RET(HcpuGetCommByGroup(group, hcclComm));

    u32 selfRank = 0;
    CHK_RET(GetGroupRankInfo(group, RankInfoType::RANK_ID_IN_GROUP, 0, &selfRank));
    std::string opTag;
    GenerateHcpuSendRecvOpTag(HCOM_OP_TYPE_SEND, group, tag, selfRank, peerRank, opTag);

    /* 接口交互信息日志 */
    HCCL_RUN_INFO("Entry-HcomSend:tag[%u], addr[%p], count[%llu], dataType[%s], peerRank[%u], opTag[%s], group[%s]",
        tag, addr,  count, GetDataTypeEnumStr(dataType).c_str(), peerRank, opTag.c_str(), group);

    /* 调用HCCL的send, 入参的正确性由HCCL确保 */
    HcclResult ret = (reinterpret_cast<hccl::HcclCommCpu *>(hcclComm.get()))->send(opTag, addr,
        count, dataType, peerRank, nullptr);
    CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Send][Result]hcclComm send error, tag[%u], "\
        "addr[%p], count[%llu], dataType[%s], peerRank[%u], group[%s], opTag[%s]", tag,
        addr, count, GetDataTypeEnumStr(dataType).c_str(), peerRank, group, opTag.c_str()), ret);

    /* 关键状态记录 */
    HCCL_RUN_INFO("hcom send success, cost time %lld us, tag[%u], addr[%p], count[%llu], dataType[%s], peerRank[%u], "\
        "opTag[%s], group[%s]", DURATION_US(TIME_NOW() - startut), tag, addr, count,
        GetDataTypeEnumStr(dataType).c_str(), peerRank, opTag.c_str(), group);
    return HCCL_SUCCESS;
}

HcclResult HcomReceiveByOS(void *addr, uint64_t count, HcclDataType dataType, uint32_t peerRank, uint32_t tag,
    const char *group, uint64_t flag)
{
    HcclUs startut = TIME_NOW();
    // 入参合法性校验
    CHK_PRT_RET(count == 0, HCCL_WARNING("input count is 0, return receive success"), HCCL_SUCCESS);
    CHK_PTR_NULL(addr);
    std::shared_ptr<hccl::HcclCommBase> hcclComm;
    CHK_RET(HcpuGetCommByGroup(group, hcclComm));

    u32 selfRank = 0;
    CHK_RET(GetGroupRankInfo(group, RankInfoType::RANK_ID_IN_GROUP, 0, &selfRank));
    std::string opTag;
    GenerateHcpuSendRecvOpTag(HCOM_OP_TYPE_RECV, group, tag, selfRank, peerRank, opTag);

    /* 接口交互信息日志 */
    HCCL_RUN_INFO("Entry-HcomReceive:tag[%u], addr[%p], count[%llu], dataType[%s], peerRank[%u], opTag[%s], group[%s]",
        tag, addr, count, GetDataTypeEnumStr(dataType).c_str(), peerRank, opTag.c_str(), group);

    HcclResult ret = (reinterpret_cast<hccl::HcclCommCpu *>(hcclComm.get()))->receive(opTag, addr, count,
        dataType, peerRank, nullptr);
    CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Receive][Result]hcclComm receive error, tag[%u], "\
        "addr[%p], count[%llu], dataType[%s], peerRank[%u], group[%s], opTag[%s]", tag,
        addr, count, GetDataTypeEnumStr(dataType).c_str(), peerRank, group, opTag.c_str()), ret);

    /* 关键状态记录 */
    HCCL_RUN_INFO("hcom receive success, cost time %lld us, tag[%u], addr[%p], count[%llu], dataType[%s], "\
        "peerRank[%u], opTag[%s], group[%s]", DURATION_US(TIME_NOW() - startut), tag, addr, count,
        GetDataTypeEnumStr(dataType).c_str(), peerRank, opTag.c_str(), group);
    return HCCL_SUCCESS;
}

HcclResult HcomGatherByOs(void* inputBuf, uint64_t inputCount, HcclDataType inputType, void* outputBuf,
    uint64_t outputCount, HcclDataType outputType, int root, const char *group, uint64_t flag)
{
    HcclUs startut = TIME_NOW();
    CHK_PRT_RET(inputCount == 0, HCCL_WARNING("input count is 0, return GatherByOs success"), HCCL_SUCCESS);
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    // 参数合法性校验
    CHK_PRT_RET(hcomInfo.pCommBase == nullptr, HCCL_ERROR("[GatherByOs][Result]hcomInfo.pComm is null, "\
        "please check if the initialize process is called."), HCCL_E_PTR);
    CHK_PTR_NULL(inputBuf);

    CHK_RET(QueryDestroyFlag(group));
    CHK_RET(HcomGroupRef(group));

    std::string strGroup = (group == nullptr) ? HCCL_WORLD_GROUP : group;
    std::shared_ptr<hccl::HcclCommBase> hcclComm;
    CHK_RET(HcpuGetCommByGroup(strGroup.c_str(), hcclComm));

    std::string tag;
    HcclCommCpu *hcclCommCpu = (reinterpret_cast<HcclCommCpu *>(hcclComm.get()));
    CHK_PTR_NULL(hcclCommCpu);
    CHK_RET(hcclCommCpu->GetHeterogCollTag(HcomOperationType::HCOM_OP_TYPE_GATHER, tag));

    /* 接口交互信息日志 */
    HCCL_RUN_INFO("Entry-HcomGatherByOs:tag[%s], inputPtr[%p], outputPtr[%p], inputCount[%llu]," \
        " rootRank[%d], dataType[%s], group[%s]", tag.c_str(), inputBuf, outputBuf, inputCount,
        root, GetDataTypeEnumStr(inputType).c_str(), strGroup.c_str());

    HcclResult ret = hcclCommCpu->GatherByOs(inputBuf, inputCount, inputType,
        outputBuf, outputCount, outputType, root, flag, tag);

    // 不是当前group的DestroyFlag，继续阻塞
    while (ret == HCCL_E_AGAIN && QueryDestroyFlag(group) != HCCL_E_AGAIN &&
        GetIncreSaveExecTimeOut() != NEVER_TIME_OUT) {
        ret = hcclCommCpu->GatherByOs(inputBuf, inputCount, inputType,
            outputBuf, outputCount, outputType, root, flag, tag);

        HCCL_RUN_INFO("HcomGatherByOs execute eagain, tag[%s], input_ptr[%p], output_ptr[%p], root[%d]"
            "count[%llu], data_type[%s]", tag.c_str(), inputBuf, outputBuf, root,
            inputCount, GetDataTypeEnumStr(inputType).c_str());
    }

    if (ret == HCCL_E_AGAIN) {
        HCCL_RUN_INFO("HcomGatherByOs has been cancelled, tag[%s], input_ptr[%p], output_ptr[%p], root[%d]"
            "count[%llu], data_type[%s]", tag.c_str(), inputBuf, outputBuf, root,
            inputCount, GetDataTypeEnumStr(inputType).c_str());
    }

    CHK_RET(HcomGroupUnref(group));
    CHK_RET(ret);
    HCCL_RUN_INFO("hcom gather by os success,take time [%lld]us, tag[%s], input_ptr[%p], output_ptr[%p], root[%d]"\
        "count[%llu], data_type[%s]", DURATION_US(TIME_NOW() - startut), tag.c_str(), inputBuf, outputBuf, root,
        inputCount, GetDataTypeEnumStr(inputType).c_str());
    return HCCL_SUCCESS;
}

HcclResult HcomBcastByOS(void* buf, uint64_t count, HcclDataType dataType, int root, const char *group, uint64_t flag)
{
    HcclUs startut = TIME_NOW();

    CHK_PRT_RET(count == 0, HCCL_WARNING("input count is 0, return broadcast success"), HCCL_SUCCESS);
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    CHK_PRT_RET(hcomInfo.pCommBase == nullptr, HCCL_ERROR("[Broadcast][Result]hcomInfo.pComm is null, "\
        "please check if the initialize process is called."), HCCL_E_PTR);
    CHK_PTR_NULL(buf);

    CHK_RET(QueryDestroyFlag(group));
    CHK_RET(HcomGroupRef(group));

    std::string strGroup = (group == nullptr) ? HCCL_WORLD_GROUP : group;
    std::shared_ptr<hccl::HcclCommBase> hcclComm = nullptr;
    CHK_RET(HcpuGetCommByGroup(strGroup.c_str(), hcclComm));

    std::string tag;
    HcclCommCpu *hcclCommCpu = reinterpret_cast<HcclCommCpu *>(hcclComm.get());
    CHK_PTR_NULL(hcclCommCpu);
    CHK_RET(hcclCommCpu->GetHeterogCollTag(HcomOperationType::HCOM_OP_TYPE_BROADCAST, tag));
    /* 接口交互信息日志 */
    HCCL_RUN_INFO("Entry-HcomBcastByOS: tag[%s], buf[%p], count[%llu], dataType[%s]," \
        " root[%d], group[%s], flag[%llu]", tag.c_str(), buf, count, GetDataTypeEnumStr(dataType).c_str(), root,
        strGroup.c_str(), flag);

    HcclResult ret = hcclCommCpu->BcastByOS(buf, count, dataType, root, flag, tag);

    // 不是当前group的DestroyFlag，继续阻塞
    while (ret == HCCL_E_AGAIN && QueryDestroyFlag(group) != HCCL_E_AGAIN &&
        GetIncreSaveExecTimeOut() != NEVER_TIME_OUT) {
        ret = hcclCommCpu->BcastByOS(buf, count, dataType, root, flag, tag);
        HCCL_RUN_INFO("HcomBcastByOS execute eagain, tag[%s], buf[%p], root_rank[%d]"
            "count[%llu], data_type[%s]", tag.c_str(), buf, root, count, GetDataTypeEnumStr(dataType).c_str());
    }

    if (ret == HCCL_E_AGAIN) {
        HCCL_RUN_INFO("HcomBcastByOS has been cancelled, tag[%s], buf[%p], root_rank[%d]"
            "count[%llu], data_type[%s]", tag.c_str(), buf, root, count, GetDataTypeEnumStr(dataType).c_str());
    }

    CHK_RET(HcomGroupUnref(group));
    CHK_RET(ret);
    HCCL_RUN_INFO("HcomBcastByOS success,take time [%lld]us, tag[%s], buf[%p], root_rank[%d]" \
        "count[%llu], data_type[%s]", DURATION_US(TIME_NOW() - startut), tag.c_str(), buf, root, count,
        GetDataTypeEnumStr(dataType).c_str());
    return HCCL_SUCCESS;
}

static HcclResult HcomInitCpuComm(const char *clusterInfoMem, const char *identify)
{
    HcclResult ret = HCCL_SUCCESS;
    bool &isAutoTuneModeOpen = HcomGetCtxAutoTuneMode();
    isAutoTuneModeOpen = false;
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();

    /*--------------入参合法性检测---------------------*/
    CHK_PTR_NULL(clusterInfoMem);
    CHK_PTR_NULL(identify);

    /* 防止重复调用初始化 */
    CHK_PRT_RET((hcomInfo.pComm != nullptr|| hcomInfo.pCommBase != nullptr),
        HCCL_ERROR("[Init][Result]errNo[0x%016llx] identify[%s],"
        "multiple initialization is not supported", HCOM_ERROR_CODE(HCCL_E_UNAVAIL), identify), HCCL_E_UNAVAIL);

    /* --------------初始化------------------------- */
    bool errorFlag = false;
    s32 logicDevId = 0;
    hcomInfo.params.commWorkMode = HCCL_MODE_SCHED_OS;
    do {
        ret = InitHcomMiscInfo(hcomInfo.params, clusterInfoMem);
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][Result]errNo[0x%016llx] init other Info.",
            HCOM_ERROR_CODE(ret)), errorFlag = true);

        DevType deviceType = DevType::DEV_TYPE_NOSOC;
        ret = CfgGetClusterInfo(clusterInfoMem, identify, hcomInfo.params, hcomInfo.rankTable, deviceType);
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][Result]errNo[0x%016llx] cfg get ranktable[%p] info "
            "error: identify[%s]", HCOM_ERROR_CODE(ret), clusterInfoMem, identify), errorFlag = true);

        const char *group;
        {
            // work mode = cpu comm固定为SCHED_BY_OS
            hcomInfo.pCommBase.reset((new (std::nothrow) hccl::HcclCommCpu(0, 0, identify)));
            CHK_PRT_RET(hcomInfo.pCommBase == nullptr, HCCL_ERROR("[Init][Result]hcomInfo.pCommBase is null,"
                "create failed"), HCCL_E_PTR);
            ret = hcomInfo.pCommBase->init(hcomInfo.params, hcomInfo.rankTable);
            CHK_PRT_BREAK(ret != HCCL_SUCCESS,
                HCCL_ERROR("[Init][Result]errNo[0x%016llx] hcclComm init error",
                HCOM_ERROR_CODE(ret)), errorFlag = true);

            group = hcomInfo.pCommBase->GetIdentifier().c_str();
        }

        ret = ShowRanktableConfigInfo(hcomInfo.cloudFlag, hcomInfo.params, hcomInfo.rankTable);
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][Result]errNo[0x%016llx] put ranktable info error",
            HCOM_ERROR_CODE(ret)), errorFlag = true);

        ret = HcomFlushBackloggedGroups();
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][Result]errNo[0x%016llx] create backlogged group failed",
            HCOM_ERROR_CODE(ret)), errorFlag = true);

        ret = HcomSetGroupTopoInfo(hcomInfo.pCommBase->GetIdentifier().c_str(), hcomInfo.rankTable.rankNum);
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][Result]errNo[0x%016llx] SetGroupTopoInfo error, "
            "group[%s]", HCOM_ERROR_CODE(ret), group), errorFlag = true);
    } while (0);

    if (errorFlag) {
        HCCL_ERROR("[Init][Result]hcom init failed, rankNum[%u], rank[%u], server[%s], device[%d], return[0x%016llx]",
            hcomInfo.rankTable.rankNum, hcomInfo.params.rank, hcomInfo.params.serverId.c_str(),
            logicDevId, HCOM_ERROR_CODE(ret));
        (void)HcomDestroy();
        return ret;
    }

    return HCCL_SUCCESS;
}


static HcclResult HcomInitCpuCommByString(const char *clusterInfoMem, const char *identify, s32 timeout)
{
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    CHK_PTR_NULL(clusterInfoMem);
    CHK_PTR_NULL(identify);

    HcclUs startut = TIME_NOW();
    HcclResult ret = HCCL_SUCCESS;
    CHK_RET(InitExternalInput());
    CHK_RET(HcomCheckInitClusterInfo(clusterInfoMem, identify));
    HCCL_RUN_INFO("Entry-HcomInitByString, identify[%s]", identify);

    ret = HcomInitCpuComm(clusterInfoMem, identify);
    CHK_RET(SetIncreSaveExecTimeOut(timeout));

    CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[HcomInitByString]errNo[0x%016llx] rankTable[%p] identify[%s] "
        "hcom init failed.", HCCL_ERROR_CODE(ret), clusterInfoMem, identify), ret);
    hcomInfo.isHcomInit = true;

    /* 关键状态记录 */
    HCCL_RUN_INFO("[HCCL_TRACE]hcom init by string success,take time [%lld]us, rankTableAddr[%p], rankNum[%u],"
        "rank[%u]", DURATION_US(TIME_NOW() - startut), clusterInfoMem, hcomInfo.rankTable.rankNum, hcomInfo.params.rank);
    return HCCL_SUCCESS;
}

HcclResult HcclCpuCommInitClusterInfoMemConfig(const char* clusterInfoMem, uint32_t rank,
    HcclCommConfig* config)
{
    CHK_PTR_NULL(config);
    CHK_PTR_NULL(clusterInfoMem);

    s32 bufferSize = sizeof(config->reserved) / sizeof(config->reserved[0]);
    s32 timeout = *(reinterpret_cast<s32 *>(&config->reserved[bufferSize - LAST_FOUR_BYTES]));
    return HcomInitCpuCommByString(clusterInfoMem, std::to_string(rank).c_str(), timeout);
}